﻿#include "IM_Socket_Server.h"
#include "Base_MessageCore.h"
#define _DEBUG
#include "Log.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/tcp.h>
#include <unistd.h>
#include <errno.h>
#include <stdlib.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdint.h>
#include <assert.h>
#include <cstring>
#include <arpa/inet.h>
#include "Atomic.h"
#include "Base_Context.h"
#include "Pri_Socket.h"
#include <functional>
#include "IM_MessageDef.h"

union sockaddr_all
{
	sockaddr s;
	sockaddr_in v4;
};


static void forward_message(int type, SocketMessage & smsg)
{
	IM_SocketMessage * sm = new IM_SocketMessage();
	sm->type = type;
	sm->ud = smsg.ud;
	sm->socket_id = smsg.socket_id;
	sm->buffer = smsg.data;

	Base_MessageCore::get_instance()->send_msg(IM_MessageType_Socket,0,smsg.context_handle, sm,sizeof(*sm));
}


static void block_readpipe(int pipefd, void *buffer, int sz)
{
	while (true)
	{
		ssize_t n = read(pipefd, buffer, sz);
		if (n < 0)
		{
			if (errno == EINTR)
			{
				continue;
			}
			fprintf(stderr, "socket_server : read pipe error %s .\n", strerror(errno));
			return;
		}


		assert(n == sz);
		if (n==sz)
		{
			return;
		}
	}
}


static int bind_socket(const char * host, uint16_t port,int protocol)
{
	assert(protocol == SocketProtocol_Tcp || protocol == SocketProtocol_Udp);

	if (host == nullptr || host[0] == 0)
	{
		host = "0.0.0.0";
	}

	sockaddr_in addr;
	addr.sin_addr.s_addr = inet_addr(host);
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	memset(&(addr.sin_zero), 0, 8);

	int fd;
	if (protocol == SocketProtocol_Tcp)
	{
		fd = socket(AF_INET, SOCK_STREAM, 0);
	}
	else
	{
		fd = socket(AF_INET, SOCK_DGRAM, 0);
	}

	if (fd > 0)
	{
		int reuse = 1;
		if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(int)) != -1)
		{
			int status = bind(fd, (sockaddr*)&addr, sizeof(sockaddr));
			if (status == 0)
			{
				return fd;
			}
		}
		close(fd);
		goto err;
	}

err:
	DEBUG("创建socket 失败！\n");
	return -1;
}

static int listen_socket(const char * host, int prot, int backlog)
{
	int listen_fd = bind_socket(host, prot,SocketProtocol_Tcp);
	if (listen_fd < 0)
	{
		return -1;
	}
	if (listen(listen_fd, backlog) == -1)
	{
		close(listen_fd);
		return -1;
	}
	return listen_fd;
}


/************************************************************************/
/*                                                                      */
/************************************************************************/



/************************************************************************/
/*                                                                      */
/************************************************************************/

#define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)

IM_Socket_Server::IM_Socket_Server()
{
}


IM_Socket_Server::~IM_Socket_Server()
{
}

bool IM_Socket_Server::socket_server_start()
{
	if (!init())
	{
		return false;
	}

	
	m_thread = new std::thread{ std::bind(&IM_Socket_Server::_socket_thread_proc,this) };

	LOG("socket 服务开始工作\n"); 
	return true;
}

void IM_Socket_Server::socket_server_close()
{
	delete m_thread;
}

bool IM_Socket_Server::socket_server_send(int socket_id, const char* buffer, int size)
{
	Pri_Socket *s = m_sockets[HASH_ID(socket_id)];
	if (s->m_socket_sate_type!=SocketStateType::SOCKET_TYPE_CONNECTED)
	{
		delete[]buffer;
		return false;
	}

	if (s->socket_direct_send((char *)buffer,size))
	{
		DEBUG("直接发送成功\n");
		return true;
	}

	struct RequestPackage request;
	request.u.send.id = socket_id;
	request.u.send.size = size;
	request.u.send.buffer = (char *)buffer;

	DEBUG("发出发送数据请求 \n");
	_send_request(request, 'D', sizeof(request.u.send));

	return true;
}

int IM_Socket_Server::sokcet_server_listen(unsigned int handle, const char *addr, int port, int backlog)
{
	int fd = listen_socket(addr, port, backlog);
	DEBUG("listen socket : %d\n",fd);
	if (fd < 0)
	{
		return -1;
	}
	RequestPackage request;
	int id = this->reserve_id();
	if (id < 0)
	{
		close(fd);
		return -1;
	}
	request.u.listen.handle = handle;
	request.u.listen.id = id;
	request.u.listen.fd = fd;
	_send_request(request, 'L', sizeof(request.u.listen));
	return id;
}

bool IM_Socket_Server::init()
{
	int fd[2];
	m_epoll = new IM_Epoll();
	
	if (pipe(fd))
	{
		delete m_epoll;
		fprintf(stderr, "socket-server: create socket pair failed.\n");
		return false;
	}

	if (!m_epoll->add_socket(fd[0], nullptr))
	{
		// add recvctrl_fd to event poll
		fprintf(stderr, "socket-server: can't add server fd to event pool.\n");
		close(fd[0]);
		close(fd[1]);
		delete m_epoll;
		return NULL;
	}

	m_event_count = 0;
	m_event_index = 0;
	m_pip_read_fd = fd[0];
	m_pip_write_fd = fd[1];
	m_check_request = true;

	for (int i=0;i<MAX_SOCKET;i++)
	{
		m_sockets[i] = nullptr;
	}

	m_alloc_id = 0;

	FD_ZERO(&m_rfds);

	assert(m_pip_read_fd < FD_SETSIZE);

	return true;
}

void IM_Socket_Server::_socket_thread_proc()
{

	while (1)
	{
		int r = _socket_poll();
		if (r<0)
		{
			break;
		}
	}
	LOG("socket server thread quit \n");
}

//返回值   -1表示失败
int IM_Socket_Server::_socket_poll()
{
	SocketMessage sokcet_message;
	int more = 1;

	int type = _socket_server_poll(sokcet_message, more);
	
	if (type==-1)
	{
		//错误
		return -1;
	}
	
	switch (type)
	{
		case SOCKET_ACCEPT:
			forward_message(SOCKET_ACCEPT,sokcet_message);
			break;
		case SOCKET_DATA:
			forward_message(SOCKET_DATA, sokcet_message);
		default:
			break;
	}
	
	if (more)
	{
		return 1;
	}
	return 1;
}

//返回值   socket 的 状态    -1 表示失败
int IM_Socket_Server::_socket_server_poll(SocketMessage& result, int &more)
{
	while (true)
	{
		//检查是否需要处理请求
		//DEBUG("检查是否需要处理请求\n");
		if (m_check_request)
		{
			if (_has_request())
			{
				DEBUG("_has_request == true \n");
				int type = _handler_request(result);
				if (type == -1)
				{
					_clear_closed_event(result, type);
					return type;
				}
				else
				{
					continue;
				}
			}
			else
			{
				m_check_request = false;
			}
		}

		//检测事件是否处理完 

		if (m_event_index == m_event_count)
		{
			m_event_count = m_epoll->wait(m_events, MAX_EVENT);
			m_check_request = true;
			more = 0;
			m_event_index = 0;
			if (m_event_count <= 0)
			{
				m_event_count = 0;
				if (errno == EINTR)
				{
					continue;
				}
				return -1;
			}
		}
		
		IM_Epoll::EpollEvent *e = &m_events[m_event_index++];
		Pri_Socket * s = (Pri_Socket*)e->s;
		if (s == nullptr)
		{
			continue;
		}
		
		switch (s->m_socket_sate_type)
		{
			/*//正在连接   目前弃用
			case SocketStateType::SOCKET_TYPE_CONNECTING:
				DEBUG("处理连接事件\n");
				return _report_connect(s, result);*/

			//Socket 是否在 监听状态 
			case SocketStateType::SOCKET_TYPE_LISTEN:
			{
				DEBUG("处理监听事件\n");
				int ok = _report_accept(s, result);
				if (ok > 0)
				{
					return SOCKET_ACCEPT;
				}
				if (ok < 0)
				{
					return SOCKET_ERR;
				}
			}
			break;
			default:
				if (e->read)
				{
					DEBUG("socket 可读事件 \n");
					int type;
					if (s->m_protocol == SocketProtocol_Tcp)
					{
						type = _forwar_message_tcp(s, result);
					}

					if (e->write&&type != SOCKET_CLOSE&&type != SOCKET_ERR)
					{
						e->read = false;
						--m_event_index;
					}
					if (type == -1)
					{
						break;
					}
					return type;
				}
				if (e->write)
				{
					DEBUG("socket 可写事件\n");
					int type = s->socket_send_buffer();
					if (type == -1)
					{
						break;
					}
					return type;
				}
				if (e->error)
				{
					DEBUG("socket 错误事件 \n");
					int error;
					socklen_t len = sizeof(error);
					int code = getsockopt(s->m_socket_fd, SOL_SOCKET, SO_ERROR, &error, &len);
					const char * err = NULL;
					if (code < 0) {
						err = strerror(errno);
					}
					else if (error != 0) {
						err = strerror(error);
					}
					else {
						err = "Unknown error";
					}
					_force_close(s, result);
					result.data = (char *)err;
					return SOCKET_ERR;
				}
				break;
		}

		
	}
}

void IM_Socket_Server::_force_close(Pri_Socket * s, SocketMessage & result)
{
	DEBUG("关闭socket\n");
#ifdef _DEBUG
	s->print_socket_infor();  //打印socket信息
#endif

	result.socket_id = s->m_socket_id;
	result.ud = 0;
	result.data = nullptr;
	result.context_handle = s->m_context_handle;
	
	if (1)
	{
		m_epoll->del_socket(s->m_socket_fd);
	}

	m_mutex.lock();
	if (1)
	{
		if (close(s->m_socket_fd))
		{
			perror("close socket: ");
		}
	}
	delete s;
	m_mutex.unlock();

}

int IM_Socket_Server::_forwar_message_tcp(Pri_Socket *s, SocketMessage &result)
{
	DEBUG("读取\n");
	int sz = s->m_read_size;
	char * buffer = new char[sz];
	int n = (int)read(s->m_socket_fd, buffer, sz);
	if (n < 0)
	{
		delete[] buffer;
		switch (errno)
		{
			case EINTR:
				break;
			case EAGAIN:
				fprintf(stderr, "socket-server: EAGAIN capture.\n");
			default:
				_force_close(s, result);
				result.data = strerror(errno);
				return SOCKET_ERR;
		}
		return -1;
	}

	if (n == 0)
	{
		delete[] buffer;
		_force_close(s, result);
		return SOCKET_CLOSE;
	}

	if (s->m_socket_sate_type== SocketStateType::SOCKET_TYPE_HALFCLOSE)
	{
		delete[] buffer;
		return -1;
	}

	result.context_handle = s->m_context_handle;
	result.socket_id = s->m_socket_id;
	result.ud = n;
	result.data = buffer;

	return SOCKET_DATA;
}



Pri_Socket * IM_Socket_Server::create_sokcet(unsigned int context_handle, int protocol, int socket_fd, bool add)
{
	int id = reserve_id();
	if (id<0)
	{
		return nullptr;
	}
	return create_sokcet(id, context_handle, protocol, socket_fd, add);
}

Pri_Socket * IM_Socket_Server::create_sokcet(int socket_id, unsigned int context_handle, int protocol, int socket_fd, bool add)
{
	Pri_Socket * s = new Pri_Socket(socket_id, context_handle, protocol, socket_fd);
	if (s)
	{
		if (add)
		{
			if (!m_epoll->add_socket(s->m_socket_fd, s))
			{
				DEBUG("添加 socket 失败 ！\n");
				assert(false);
				m_sockets[socket_id] = nullptr;
				delete s;
				return nullptr;
			}
		}
		m_sockets[socket_id] = s;
		return s;
	}
	return nullptr;
}

int IM_Socket_Server::_report_connect(Pri_Socket * s, SocketMessage & result)
{
	int error;
	socklen_t soclet_len = sizeof(error);
	int code = getsockopt(s->m_socket_fd, SOL_SOCKET, SO_ERROR, &error, &soclet_len);
	if (code < 0 || error)
	{
		_force_close(s, result);
		if (code >= 0)
		{
			result.data = strerror(error);
		}
		else
		{
			result.data = strerror(error);
		}
		return SOCKET_ERR;
	}
	else
	{
		s->m_socket_sate_type=SOCKET_TYPE_CONNECTED;
		result.context_handle = s->m_context_handle;
		result.socket_id = s->m_socket_id;
		result.ud = 0;
		result.data = nullptr;
		if (s->nomore_send_data())
		{
			m_epoll->set_write_socket(s->m_socket_fd, s, false);
		}

		return SOCKET_OPEN;
	}
}

int IM_Socket_Server::reserve_id()
{
	
	for (int i = 0; i < MAX_SOCKET; i++)
	{
		int id = ATOM_INC(&this->m_alloc_id);
		if (id < 0)
		{
			id = ATOM_AND(&this->m_alloc_id, 0x7fffffff);
		}
		Pri_Socket * s = this->m_sockets[HASH_ID(id)];
		if (s==nullptr)
		{
			LOG("分配一个socket id： %d\n", id);
			return id;
		}
	}
	return -1;
}

int IM_Socket_Server::_report_accept(Pri_Socket * s, SocketMessage & result)
{
	sockaddr_all u;
	socklen_t len = sizeof(u);
	int client_fd = accept(s->m_socket_fd, &u.s, &len);
	if (client_fd < 0)
	{
		if (errno == EMFILE || errno == ENFILE)
		{
			result.context_handle = s->m_context_handle;
			result.socket_id = s->m_socket_id;
			result.ud= 0;
			result.data = strerror(errno);
			return -1;
		}
		else
		{
			return 0;
		}
	}

	Pri_Socket * ss =create_sokcet(s->m_context_handle,SocketProtocol_Tcp,client_fd,true);
	if (ss==nullptr)
	{
		return -1;
	}
	ss->set_secket_keepalive();
	ss->set_socket_noblocking();

	ss->m_socket_sate_type=SocketStateType::SOCKET_TYPE_CONNECTED;

	result.context_handle = s->m_context_handle;
	result.socket_id = s->m_socket_id;
	result.ud = ss->m_socket_id;
	result.data = nullptr;

	return 1;
}

bool IM_Socket_Server::_has_request()
{
	timeval tv{ 0,0 };
	int retval;
	FD_SET(m_pip_read_fd, &m_rfds);

	retval = select(m_pip_read_fd + 1, &m_rfds, nullptr, nullptr,&tv);

	if (retval==1)
	{
		return true;
	}
	return false;
}

int IM_Socket_Server::_handler_request(SocketMessage & result)
{
	int fd = m_pip_read_fd;
	uint8_t buffer[256];
	uint8_t header[2];
	block_readpipe(fd, header, sizeof(header));
	int type = header[0];
	int len = header[1];
	block_readpipe(fd, buffer, len);
	switch (type)
	{
		//发送数据
		case 'D':
			return _socket_send((Request_send *)buffer, result);
		case 'L':
			return _socket_listen((RequestListen *)buffer, result);
		default:
			break;
	}
	return -1;
}

void IM_Socket_Server::_send_request(RequestPackage & request, char type, int len)
{
	request.header[6] = (uint8_t)type;
	request.header[7] = (uint8_t)len;
	while (true)
	{
		ssize_t n = write(this->m_pip_write_fd, &request.header[6], len + 2);
		if (n < 0)
		{
			if (errno != EINTR)
			{
				//
			}
		}
		assert(n == len + 2);
		DEBUG("发出 发送数据请求 \n");
		return;
	}
}

/*
处理发送消息的请求

如果 socket 有效  则  加入发送队列  
*/
int IM_Socket_Server::_socket_send(Request_send* request, SocketMessage & result)
{
	DEBUG("处理发送请求\n");
	int id = request->id;
	Pri_Socket * s = m_sockets[HASH_ID(id)];
	
	if (s->m_socket_sate_type==SOCKET_TYPE_HALFCLOSE)
	{
		delete[] request->buffer;
		return -1;
	}

	if (s->m_socket_sate_type==SOCKET_TYPE_CONNECTED)
	{
		if (s->m_protocol==SocketProtocol_Tcp)
		{
			s->push_data_to_write_buffer(request->buffer, request->size);

			m_epoll->set_write_socket(s->m_socket_fd, s, true);

			DEBUG("处理发送请求完成\n");
			return 0;
		}
		
	}
	return -1;
}

int IM_Socket_Server::_socket_listen(RequestListen * request, SocketMessage & result)
{
	int id = request->id;
	int listen_fd = request->fd;
	Pri_Socket* s= create_sokcet(id, request->handle, SocketProtocol_Tcp, listen_fd, true);
	DEBUG("socket mmaddr : %p \n", s);
	if (s)
	{
		s->m_socket_sate_type = SocketStateType::SOCKET_TYPE_LISTEN;
		return id;
	}
	else
	{
		close(listen_fd);
		result.context_handle = request->handle;
		result.socket_id = id;
		result.ud = 0;
		result.data = "reach socket number limit";
		m_sockets[HASH_ID(id)] = nullptr;
	}
	return -1;
}

void IM_Socket_Server::_clear_closed_event(SocketMessage&result, int type)
{
	if (type==SOCKET_CLOSE||type==SOCKET_ERR)
	{
		int id = result.socket_id;
		for (int i=m_event_index;i<m_event_count;i++)
		{
			IM_Epoll::EpollEvent & e = m_events[i];
			Pri_Socket * s = (Pri_Socket *)e.s;
			if (s)
			{
				if (s->m_socket_id==id)
				{
					e.s = nullptr;
				}
			}
		}
	}
}

